package com.frinder.official.akka.routing;

import akka.actor.ActorRef;
import akka.routing.ConsistentHashingPool;
import akka.routing.DefaultResizer;
import com.frinder.official.spring.MyBaseActor;
import com.frinder.official.spring.SpringExtension;
import com.frinder.official.utils.Utils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import static akka.routing.ConsistentHashingRouter.ConsistentHashMapper;

/**
 * Created by frinder on 2017/9/29.
 */
@Component
@Scope("prototype")
public class ConsistentHashingRouter extends MyBaseActor {

    final double goldenSectionPoint = 0.618;
    final int routees = 5;

    final DefaultResizer resizer = new DefaultResizer(
            (int) Math.round(routees * goldenSectionPoint),
            (int) Math.round(routees * (1 + goldenSectionPoint))
    );

    final ConsistentHashMapper hashMapper = (Object message) -> {
        if (message instanceof PayloadActor.Payload) {
            return Utils.remainder(((PayloadActor.Payload) message).getKey(), routees);
        } else {
            return routees;
        }
    };

    @Autowired
    SpringExtension springExtension;

    ActorRef router;

    @Override
    public void preStart() throws Exception {
        super.preStart();
        router = getContext().actorOf(
                new ConsistentHashingPool(routees)
                        .withResizer(resizer)
                        .withHashMapper(hashMapper)
                        .props(springExtension.props("payloadActor")),
                "consistent-hashing-router"
        );
    }

    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(PayloadActor.Payload.class, payload -> router.tell(payload, getSender()))
                .build();
    }
}
